package com.sdses.flink.sql;

import com.sdses.flink.udf.PgGeometryTypeConvert;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkSQL_HgCDC2HgTest {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger(RestOptions.PORT, 18081);
        conf.setString(RestOptions.BIND_PORT, "18081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.createTemporaryFunction("geometry_type_convert", PgGeometryTypeConvert.class);
        //SourceTable 瀚高
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    code STRING ,\n" +
                "    name STRING,\n" +
                "    id BIGINT,\n" +
                "    shape_length FLOAT,\n" +
                "    shape_area FLOAT,\n" +
                "    the_geom STRING,\n" +
//                "    import_time_ TIMESTAMP \n" +
                "    import_time_ TIMESTAMP, \n" +
                "    create_date DATE\n" +
                ") WITH (\n" +
                "  'connector' = 'postgres-cdc',\n" +
                "  'hostname' = '192.168.102.154',\n" +
                "  'port' = '5866',\n" +
                "  'username' = 'sysdba',\n" +
                "  'password' = 'Sdses,hg2o23',\n" +
                "  'database-name' = 'db_datacube_sdses_test',\n" +
                "  'schema-name' = 'public',\n" +
                //TODO 解决瀚高Date类型报错问题，牺牲毫秒以下精度为代价，并且引入时区问题
                "  'debezium.time.precision.mode' = 'connect',\n" +
//                "  'table-name' = 'test_gis_erp_community',\n" +
                "  'table-name' = 'test_gis_erp_community2',\n" +
                "  'slot.name' = 'flinksql',\n" +
                "  'decoding.plugin.name' = 'pgoutput'\n" +
                ");");

        //SinkTable 瀚高
        tableEnv.executeSql("CREATE TABLE sinkTable (\n" +
                "    code STRING ,\n" +
                "    name STRING,\n" +
                "    id BIGINT,\n" +
                "    shape_length FLOAT,\n" +
                "    shape_area FLOAT,\n" +
                "    the_geom STRING,\n" +
                "    import_time_ TIMESTAMP, \n" +
                "    create_date DATE, \n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'jdbc',\n" +
                "  'url' = 'jdbc:postgresql://192.168.102.154:5866/db_datacube_sdses_test?stringtype=unspecified',\n" +
                "  'table-name' = 'test_gis_erp_community',\n" +
                "  'username' = 'sysdba',\n" +
                "  'password' = 'Sdses,hg2o23'\n" +
                ");");

        //SinkTable Print
//        tableEnv.executeSql("CREATE TABLE sinkTable (\n" +
//                "    code STRING ,\n" +
//                "    name STRING,\n" +
//                "    id BIGINT,\n" +
//                "    shape_length FLOAT,\n" +
//                "    shape_area FLOAT,\n" +
//                "    the_geom STRING,\n" +
//                "    import_time_ TIMESTAMP, \n" +
//                "    create_date STRING, \n" +
//                "    PRIMARY KEY (id) NOT ENFORCED\n" +
//                ") WITH (\n" +
//                "  'connector' = 'print' \n" +
//                ");");


//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");


//        tableEnv.executeSql("select code, name, id, shape_length, shape_area, the_geom, import_time_, create_date from sourceTable")
//                        .print();

        tableEnv.executeSql("insert into sinkTable select code, name, id, shape_length, shape_area, geometry_type_convert(the_geom), import_time_, create_date from sourceTable");


    }

}
